package com.li.kafka.admin;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.config.ConfigResource;

import java.util.*;
import java.util.concurrent.ExecutionException;

public class AdminSample {
  public static AdminClient adminClient;
  public static final String TOPIC_NAME = "topic_example";

  public static void main(String[] args) throws ExecutionException, InterruptedException {
    //    AdminSample.create();
    //    AdminSample.delete();
    //    AdminSample.topics();
    //    AdminSample.describe();
    //    AdminSample.alterConfig();
    //    AdminSample.describeConfig();
    //    AdminSample.incrPartitions(2);
    //    AdminSample.describe();
  }

  /** 增加分区 */
  public static void incrPartitions(int partitions) {
    AdminClient client = AdminSample.adminClient();
    Map<String, NewPartitions> map = new HashMap<>();
    NewPartitions newPartitions = NewPartitions.increaseTo(partitions);
    map.put(TOPIC_NAME, newPartitions);
    client.createPartitions(map);
  }

  public static void alterConfig() {
    AdminClient client = AdminSample.adminClient();
    Map<ConfigResource, Config> cfgMaps = new HashMap<>();

    ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME);
    Config cfg = new Config(Arrays.asList(new ConfigEntry("preallocate", "true")));

    cfgMaps.put(configResource, cfg);
    AlterConfigsResult alterConfigsResult = client.alterConfigs(cfgMaps);
  }

  /** Topic 描述 */
  public static void describe() throws ExecutionException, InterruptedException {
    AdminClient client = AdminSample.adminClient();
    DescribeTopicsResult describeTopicsResult = client.describeTopics(Arrays.asList(TOPIC_NAME));
    Map<String, TopicDescription> stringTopicDescriptionMap = describeTopicsResult.all().get();
    Set<Map.Entry<String, TopicDescription>> entries = stringTopicDescriptionMap.entrySet();
    entries.stream()
        .forEach(
            (entry) -> {
              System.out.println(entry.getKey() + ":" + entry.getValue());
            });
  }
  /** Topic 配置信息查看 */
  public static void describeConfig() throws ExecutionException, InterruptedException {
    AdminClient client = AdminSample.adminClient();
    ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME);
    DescribeConfigsResult describeConfigsResult =
        client.describeConfigs(Arrays.asList(configResource));
    Map<ConfigResource, Config> configResourceConfigMap = describeConfigsResult.all().get();
    Set<Map.Entry<ConfigResource, Config>> entries = configResourceConfigMap.entrySet();
    entries.stream()
        .forEach(
            (entry) -> {
              System.out.println(entry.getKey() + ":" + entry.getValue());
            });
  }

  /** 创建Topic */
  public static void create() {
    AdminClient client = AdminSample.adminClient();
    Short rs = 1; // 副本因子
    NewTopic topic = new NewTopic(TOPIC_NAME, 1, rs);
    CreateTopicsResult result = client.createTopics(Arrays.asList(topic));
    System.out.println(result);
  }

  /** 删除Topic */
  public static void delete() {
    AdminClient client = AdminSample.adminClient();
    DeleteTopicsResult result = client.deleteTopics(Arrays.asList(TOPIC_NAME));
    System.out.println(result);
  }

  /** 查看Topic集合 */
  public static void topics() throws ExecutionException, InterruptedException {
    AdminClient client = AdminSample.adminClient();
    ListTopicsOptions options = new ListTopicsOptions();
    options.listInternal(true);
    ListTopicsResult list = client.listTopics(options);
    Set<String> names = list.names().get();
    names.stream().forEach(System.out::println);
  }

  public static AdminClient adminClient() {
    if (AdminSample.adminClient == null) {
      Properties prop = new Properties();
      prop.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "1.15.61.173:9092");
      AdminSample.adminClient = AdminClient.create(prop);
    }
    return AdminSample.adminClient;
  }
}
